#pragma once

// 异步工作器

#include "buffer.hpp"
#include <mutex>
#include <condition_variable>
#include <functional>
#include <memory>
#include <thread>
#include <atomic>

namespace jiuqilog
{
    using Func = std::function<void(Buffer &)>;
    enum class AsyncType
    {
        ASYNC_SAFE,  // 缓冲区满了就阻塞
        ASYNC_UNSAFE // 无限扩容
    };
    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        AsyncLooper(const Func &callback, AsyncType looper_type = AsyncType::ASYNC_SAFE)
            : _stop(false),
              _thread(std::thread(&AsyncLooper::threadEntry, this)),
              _callback(callback),
              _looper_type(looper_type) {}

        void push(const char *data, size_t len)
        {
            // 1.无限扩容--非安全
            // 2.固定大小,生产缓冲区数据满了就阻塞
            std::unique_lock<std::mutex> lock(_mutex);
            if (_looper_type == AsyncType::ASYNC_SAFE)
                _cond_pro.wait(lock, [&]()
                               { return _pro_buf.writeAbleSize() >= len; }); //如果是安全的且没有足够空间就阻塞
            // 走下来代表可以向缓冲区添加数据
            _pro_buf.push(data, len);
            _cond_con.notify_one(); // 唤醒日志输出线程处理数据
        }

        void stop()
        {
            _stop = true;           // 设置退出标志为true
            _cond_con.notify_all(); // 唤醒所有工作线程,处理剩余数据后退出
            _thread.join();
        }

        ~AsyncLooper()
        {
            stop();
        }

    private:
        // 线程入口函数
        void threadEntry()
        {
            while (1)
            {
                // 如果生产缓冲区有数据,就取出来,没有就阻塞
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    if (_stop && _pro_buf.empty())
                        break;
                    _cond_con.wait(lock, [&]()
                                   { return _stop || !_pro_buf.empty(); });
                    _con_buf.swap(_pro_buf);
                    // 唤醒生产者(安全状态下)
                    if (_looper_type == AsyncType::ASYNC_SAFE)
                        _cond_pro.notify_all();
                }
                // 处理消费缓冲区的数据
                _callback(_con_buf);
                // 初始化消费缓冲区
                _con_buf.reset();
            }
        }

    private:
        Func _callback;

    private:
        std::atomic<bool> _stop; // 停止工作标志
        Buffer _pro_buf;         // 生产缓冲区
        Buffer _con_buf;         // 消费缓冲区
        std::mutex _mutex;
        std::condition_variable _cond_pro;
        std::condition_variable _cond_con;
        std::thread _thread; // 异步工作线程
        AsyncType _looper_type;
    };
}